package org.rency.common.mq.handler.impl;

import org.rency.common.mq.enums.DestinationType;
import org.rency.common.mq.failover.FailoverService;
import org.rency.common.mq.handler.MessageHandler;
import org.rency.common.mq.request.MQRequest;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public abstract class AbstractHandler implements MessageHandler {

	private static final Logger logger = LoggerFactory.getLogger(AbstractHandler.class);
	private static final String DEFAULT_RETRY_QUEUE = "org.rency.mq.retry.queue";
	private static final String DEFAULT_FAILED_QUEUE = "org.rency.mq.failed.queue";
	private static final int DEFAULT_MAX_RETRY_TIMES = 3;

	private FailoverService failoverService;
	/**
	 * 并发消费者数量
	 */
	private int concurrentConsumers = 1;
	private String retryQueue = DEFAULT_RETRY_QUEUE;
	private String failedQueue = DEFAULT_FAILED_QUEUE;
	private int retryTimes = DEFAULT_MAX_RETRY_TIMES;
	
	@Override
	public void handler(Object request) {
		logger.debug("Handle message: " + request);
		MQRequest mqRequest = convertToMQRequest(request);

        try {
            if (mqRequest == null) {
                handleMessage(request);
                return;
            }

            Object content = getMessageContent(mqRequest);

            if (content == null) {
            	logger.warn("The content is null. Message: " + request);
                return;
            }
            logger.info("Message content: " + content);
            handleMessage(content);
        } catch (Throwable e) {
        	logger.warn("Handling message failed, and retry it...", e);
            //重试处理失败的消息
            if (mqRequest != null) {
            	if(mqRequest.getDestinationType() == DestinationType.TOPIC){
            		logger.warn("Topic messages cannot retry." + mqRequest);
            	}else{
	    			if (!mqRequest.containsProperty("retryQueue")) {
	    				mqRequest.addProperty("retryQueue", retryQueue);
	    				mqRequest.addProperty("retryTimes", retryTimes);
	    				mqRequest.addProperty("failedQueue", failedQueue);
	    				mqRequest.addProperty("failedTimes", 0);
                        mqRequest.addProperty("_exception", e);
	    			}

	    			getFailoverService().retry(mqRequest);
            	}
    		}else{
    			getFailoverService().retry(request, retryTimes, retryQueue, failedQueue);
    		}
        }
	}

	
	/**
     * 获取MQRequest的消息内容
     * @param mqRequest  MQRequest对象
     * @return
     */
    protected Object getMessageContent(MQRequest request) {
        return request;
    }
	
	public FailoverService getFailoverService() {
		return failoverService;
	}

	public void setFailoverService(FailoverService failoverService) {
		this.failoverService = failoverService;
	}

	public int getConcurrentConsumers() {
		return concurrentConsumers;
	}

	public void setConcurrentConsumers(int concurrentConsumers) {
		this.concurrentConsumers = concurrentConsumers;
	}

	public String getRetryQueue() {
		return retryQueue;
	}

	public void setRetryQueue(String retryQueue) {
		this.retryQueue = retryQueue;
	}

	public String getFailedQueue() {
		return failedQueue;
	}

	public void setFailedQueue(String failedQueue) {
		this.failedQueue = failedQueue;
	}

	public int getRetryTimes() {
		return retryTimes;
	}

	public void setRetryTimes(int retryTimes) {
		this.retryTimes = retryTimes;
	}

}
